package com.arelikebrothers.demo.kafka.stream;

import com.google.gson.Gson;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

/**
 * Created by lennon on 16/07/2017.
 */
public class JsonDeserializer<T> implements Deserializer<T> {
    public static final String DESERIALIZER_KEY = "JsonDeserializer";
    private Class<T> clsType;

    private Gson gson = new Gson();

    @Override
    public void configure(Map<String, ?> map, boolean b) {
        clsType = (Class<T>) map.get(DESERIALIZER_KEY);
    }

    @Override
    public T deserialize(String s, byte[] bytes) {
        if (bytes == null) {
            return null;
        }

        String jsonString = new String(bytes);
        T obj;
        try {
            obj = gson.fromJson(jsonString, clsType);
            return obj;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void close() {

    }
}
